/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;

import lib.Tuple;

import com.hp.hpl.jena.clusteredtdb.comms.ClusterRepresentation;
import com.hp.hpl.jena.clusteredtdb.pgraph.NetworkedGraphTDBFactory;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.sparql.util.NodeFactory;
import com.hp.hpl.jena.tdb.base.file.Location;
import com.hp.hpl.jena.tdb.store.FactoryGraphTDB;
import com.hp.hpl.jena.tdb.store.GraphTDB;
import com.hp.hpl.jena.tdb.store.NodeId;

public class MultiThreadedTest {
	
	ArrayList<String> resultSet;
	int DEBUG = 0;
	
	public MultiThreadedTest(boolean standalone, String serverConfig, int users, int maxThreadsPerUser, List<String> filebases) throws Exception {
		
		ClusteredTDBConfig ctdb = new ClusteredTDBConfig(serverConfig);
		GraphTDB gtdb;
		if(standalone) {
			gtdb = FactoryGraphTDB.createGraph(new Location(ctdb.getDataFileBase()));
		} else {
			
			ClusterRepresentation cr = new ClusterRepresentation(ctdb);
			gtdb = NetworkedGraphTDBFactory.create(new Location(ctdb.getDataFileBase()),cr,ctdb);
		}
		
		Iterator<String> fbases = filebases.iterator();
		
		while(fbases.hasNext()) {
			resultSet = new ArrayList<String>();
			ArrayList<Thread> tl = new ArrayList<Thread>();
			String filebase = fbases.next();
//			begin warmup

			System.out.println("Test system using " +users + " users with up to " + maxThreadsPerUser +" threads each on filebase " + filebase );
			System.out.println("Beginning warmup");
			User u = new User(this,gtdb,-1,filebase+".warmup",maxThreadsPerUser);
			Thread t = new Thread(u);
			t.start();
			t.join();

			if(DEBUG == 1) {
				for(int i = 0; i < this.resultSet.size(); i++) {
					System.out.println(resultSet.get(i));
				}
			}

			System.out.println("Warmup complete\n\n");
//			end warmup

			//reset resultsl.remove(0);
			resultSet = new ArrayList<String>();
//			begin tests
			long inittime = System.currentTimeMillis();
			for(int i = 0; i < users; i++) {
				String file = filebase+"."+i;
				u = new User(this,gtdb,i,file,maxThreadsPerUser);
				t = new Thread(u);
				tl.add(t);
				t.start();
			}
			Iterator<Thread> it = tl.iterator();
			while(it.hasNext()) {
				it.next().join();
			}

			for(int i = 0; i < this.resultSet.size(); i++) {
				System.out.println(resultSet.get(i));
			}

			System.out.println("Time taken for all actions: " + (System.currentTimeMillis() - inittime));
		}
	}
	
	public synchronized void addResults(ArrayList<String> al) {
		this.resultSet.addAll(al);
	}
	
	public static void main (String[] args) throws Exception {
		ArrayList<String> l = new ArrayList<String>(Arrays.asList(args)); 
		String serverConfig = "resources/serverconfig.properties";
		int maxThreadsPerUser = 1;
		int users = 1;
		boolean standalone = true;
		if(args.length < 4) {
			System.out.println("ERROR: bad args.  Use: MultiThreadedTest <cluster/standalone> <server config file> <users> <maxthreadsperuser> testfiles...");
		}
		if(l.get(0).equals("cluster")) {
			standalone = false;
		}
		serverConfig = l.get(1);
		users = Integer.parseInt(l.get(2));
		maxThreadsPerUser = Integer.parseInt(l.get(3));
		l.remove(0);
		l.remove(0);
		l.remove(0);
		l.remove(0);
		
		new MultiThreadedTest(standalone,serverConfig,users,maxThreadsPerUser,l);
	}
	
	class User implements Runnable {

		GraphTDB gtdb;
		int maxThreads;
		int userNumber;
		BufferedReader testFile;
		ArrayList<String> result;
		MultiThreadedTest parent;
		
		public User(MultiThreadedTest parent, GraphTDB gtdb, int userNumber, String file, int maxThreads) throws FileNotFoundException {
			this.parent = parent;
			this.maxThreads = maxThreads;
			this.gtdb = gtdb;
			this.testFile = new BufferedReader(new FileReader(file));
			this.result = new ArrayList<String>();
			this.userNumber = userNumber;
		}
		
		public synchronized void addResultLine(String result) {
			this.result.add("User: " + userNumber + ", " + result);
		}
		
		public void run() {
			String line;
			
			try {
				while((line = testFile.readLine()) != null) {
					if(line != "") {
						QueryExecutor qe = new QueryExecutor(this,gtdb,line,maxThreads);
						qe.execute();
					}
				}
				parent.addResults(result);
				
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	}
	
	class TripleReader implements Runnable {

		NodeId s,p,o;
		GraphTDB gtdb;
		QueryExecutor parent; 
		
		public TripleReader(QueryExecutor parent, GraphTDB gtdb, NodeId s, NodeId p, NodeId o) {
			this.s = s;
			this.p = p;
			this.o = o;
			this.gtdb = gtdb;
			this.parent = parent;
		}
		
		public void run() {
			parent.updateCount(parent.exhaustIteratorTuple(gtdb.getNodeTupleTable().find(s, p, o)));
			parent.updateThreadCount(-1);
		}
		
	}
	
	class QueryExecutor {

		String queryDescription = "";
		String queryType = "";
		GraphTDB gtdb;
		Iterator<String> queryTerms;
		HashSet<String> done;
		int maxThreads;
		int count = 0;
		int threadcount = 0;
		User parent;
		
		public QueryExecutor(User parent, GraphTDB gtdb, String queryLine, int maxThreads) {
			this.maxThreads = maxThreads;
			this.gtdb = gtdb;
			done = new HashSet<String>();
			parseQuery(queryLine);
			this.parent = parent;
		}
		
		private void parseQuery(String queryLine) {
			String[] query = queryLine.split(",");
			queryTerms = Arrays.asList(query).iterator();
			queryDescription = queryTerms.next();
			queryType = queryTerms.next();
			
		}

		public synchronized void updateCount(int add) {
			this.count += add;
		}
		
		public synchronized void updateThreadCount(int add) {
			this.threadcount += add;
			this.notify();
		}
		
		public void execute() {
			long queryInitTime = System.currentTimeMillis();
			int results = 0;
			if(queryType.equals("Join")) {
				results = join();
			} else if (queryType.equals("TripleNodes")) {
				results = exhaustIteratorTriple(tripleNodes());
			} else if (queryType.equals("TripleNoNodes")) {
				results = exhaustIteratorTuple(tripleNoNodes());
			}
			parent.addResultLine("Query: " + queryDescription + " took " + (System.currentTimeMillis() - queryInitTime) + "ms to complete, returning " + results + " results.");
		}

        private Iterator<Triple> tripleNodes() {
			Node[] nodes = getNodeTriple();
			@SuppressWarnings("unchecked")
			Iterator<Triple> it = gtdb.find(nodes[0],nodes[1],nodes[2]);
			return it;
			
		}
		
		private int exhaustIteratorTuple(Iterator<Tuple<NodeId>> it) {
			int i = 0;
			while(it.hasNext()) {
				Tuple<NodeId> tup = it.next();
				if(DEBUG > 1) {
					System.out.println(tup.get(0)+"," + tup.get(1)+"," +tup.get(2));
				}
				i++;
			}
			return i;
		}
		
		private int exhaustIteratorTriple(Iterator<Triple> it) {
			int i = 0;
			while(it.hasNext()) {
				it.next();
				i++;
			}
			return i;
		}

		private Node[] getNodeTriple() {
			Node[] nodes = new Node[3];
			for(int i = 0; i < 3; i++) {
				String temp = queryTerms.next();
				if(!temp.equals("null")) {
					nodes[i] = NodeFactory.create(temp);
				}
			}			
			return nodes;
		}
		
		//create a triple from the next three terms in the queryTerms iterator
		private NodeId[] getTripleFromNodes() {
			Node[] nodes = getNodeTriple();
			NodeId[] nids = new NodeId[3];
						
			for(int i = 0; i < 3; i++) {
				if(nodes[i] == null) {
					nids[i] = null;
				} else {
					nids[i] = gtdb.getNodeTupleTable().getNodeTable().getNodeIdForNode(nodes[i]);
				}
			}
			
			return nids;
		}

		private Iterator<Tuple<NodeId>> tripleNoNodes() {
			NodeId[] triple = getTripleFromNodes();
			Iterator<Tuple<NodeId>> it = gtdb.getNodeTupleTable().find(triple[0],triple[1], triple[2]);
			return it;
		}
		
		//er, loops, anyone?  Temp code, will refactor later if really needed.
		private int join() {
			int srep,prep,orep;
			
			srep = prep = orep = -2;
			
			String secondS = queryTerms.next();
			String secondP = queryTerms.next();
			String secondO = queryTerms.next();
			NodeId joinS, joinP, joinO;
			joinS = joinP = joinO = null;
			
			if(secondS.substring(0, 1).equals("?")) {
				srep = Integer.parseInt(secondS.substring(1, 2));
			} else if(!secondS.equals("null")) {
				joinS = gtdb.getNodeTupleTable().getNodeTable().getNodeIdForNode(NodeFactory.create(secondS));
			}
			if(secondP.substring(0, 1).equals("?")) {
				prep = Integer.parseInt(secondP.substring(1, 2));
			} else if(!secondP.equals("null")) {
				joinP = gtdb.getNodeTupleTable().getNodeTable().getNodeIdForNode(NodeFactory.create(secondP));
			}
			if(secondO.substring(0, 1).equals("?")) {
				orep = Integer.parseInt(secondO.substring(1, 2));
			} else if(!secondO.equals("null")) {
				joinO = gtdb.getNodeTupleTable().getNodeTable().getNodeIdForNode(NodeFactory.create(secondO));
			}
			
			Iterator<Tuple<NodeId>> it = tripleNoNodes();
			int debugCount = 0;
			while(it.hasNext()) {
				Tuple<NodeId> tup = it.next();
				
				NodeId s, p, o;
				s = joinS;
				p = joinP;
				o = joinO;
				
				if(srep > -1) {
					s = tup.get(srep);
				}
				if(prep > -1) {
					p = tup.get(prep);
				}
				if(orep > -1) {
					o = tup.get(orep);
				}
				this.updateCount(1);
				debugCount++;
				if(!alreadyDone(s,p,o)) {
					try {
						synchronized(this) {
							while(this.threadcount == this.maxThreads) {
								this.wait();
							}
							
							this.threadcount++;
							TripleReader tr = new TripleReader(this,gtdb,s,p,o); 
							Thread t = new Thread(tr);
							t.start();
							
						}
					} catch (InterruptedException e) {
						
					}
				}
				
			}
			if(DEBUG > 0) {
				parent.addResultLine("Results to be joined on: " + debugCount);
			}
			synchronized(this) {
				while(this.threadcount > 0) {
					try {
						this.wait();
					} catch (InterruptedException e) {
						
					}
				}
			}
			return this.count;
		}

		private boolean alreadyDone(NodeId s, NodeId p, NodeId o) {
			String trip = "";
			if(s == null) {
				trip += "<>";
			} else {
				trip += "<" + s + ">";
			}
			
			if(p == null) {
				trip += "<>";
			} else {
				trip += "<" + p + ">";
			}
			
			if(o == null) {
				trip += "<>";
			} else {
				trip += "<" + o + ">";
			}
			
			if(done.contains(trip)) {
				return true;
			} else {
				done.add(trip);
				return false;
			}
		}
		
	}
}

/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
